package cn.com.greatwall.kafka.service.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.servlet.http.HttpServletResponse;

import org.apache.commons.lang3.StringUtils;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSONObject;

import cn.com.greatwall.common.utils.FileUtil;
import cn.com.greatwall.common.utils.PageUtil;
import cn.com.greatwall.common.utils.QueryHelp;
import cn.com.greatwall.common.utils.TimeUtil;
import cn.com.greatwall.common.utils.ValidationUtil;
import cn.com.greatwall.kafka.domain.Cluster;
import cn.com.greatwall.kafka.domain.Topic;
import cn.com.greatwall.kafka.repository.TopicRepository;
import cn.com.greatwall.kafka.service.ClusterService;
import cn.com.greatwall.kafka.service.KafkaService;
import cn.com.greatwall.kafka.service.TopicService;
import cn.com.greatwall.kafka.service.dto.ClusterDto;
import cn.com.greatwall.kafka.service.dto.TaskDto;
import cn.com.greatwall.kafka.service.dto.TopicDto;
import cn.com.greatwall.kafka.service.dto.criteria.TopicQueryCriteria;
import cn.com.greatwall.kafka.service.mapstruct.TopicMapper;
import cn.com.greatwall.kafka.util.KafkaUtil;
import cn.com.greatwall.system.modules.system.domain.Dept;
import lombok.RequiredArgsConstructor;

/**
 * @Author herw
 * @Time 2021-01-21 13:29:29
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
@Service
@RequiredArgsConstructor
public class TopicServiceImpl implements TopicService {
    private final TopicRepository topicRepository;
    private final TopicMapper topicMapper;
    private final KafkaService kafkaService;
    private final ClusterService clusterService;

    @Override
    public void download(List<TopicDto> topicDtos, HttpServletResponse response) throws IOException {
        List<Map<String, Object>> list = new ArrayList<>();
        for (TopicDto topicDTO : topicDtos) {
            Map<String, Object> map = new LinkedHashMap<>();
            map.put("集群名称", topicDTO);
            list.add(map);
        }
        FileUtil.downloadExcel(list, response);
    }

    @Override
    public Map<String, Object> queryAll(TopicQueryCriteria criteria, Pageable pageable) {
        Page<Topic> page = topicRepository.findAll(
                (root, criteriaQuery, criteriaBuilder) -> QueryHelp.getPredicate(root, criteria, criteriaBuilder),
                pageable);
        return PageUtil.toPage(page.map(topicMapper::toDto));
    }

    @Override
    public List<TopicDto> queryAll(TopicQueryCriteria criteria) {
        return topicMapper.toDto(topicRepository.findAll(
                (root, criteriaQuery, criteriaBuilder) -> QueryHelp.getPredicate(root, criteria, criteriaBuilder)));
    }
    
    @Override
    public List<Topic> findByClusterId(String clusterId) {
        return topicRepository.findByClusterId(Long.parseLong(clusterId));
    }
    
    @Override
    public TopicDto findById(Long id) {
        Topic topic = topicRepository.findById(id).orElseGet(Topic::new);
        ValidationUtil.isNull(topic.getId(), "Topic", "id", id);
        return topicMapper.toDto(topic);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void create(Topic resources) {
        topicRepository.save(resources);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void update(Topic resources) {
        Topic topic = topicRepository.findById(resources.getId()).orElseGet(Topic::new);
        ValidationUtil.isNull(topic.getId(), "Topic", "id", resources.getId());
        resources.setId(topic.getId());
        topicRepository.save(resources);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void delete(Set<Long> ids) {
        topicRepository.deleteAllByIdIn(ids);
    }

    @Override
    public JSONObject encapsulatedObject(TaskDto task) throws Exception {
        JSONObject obj = new JSONObject();
        String[] clusterIds = task.getClusterIds().split(",");
        if (clusterIds.length > 0) {
            boolean flag = true;
            for (String clusterId : clusterIds) {
                if (StringUtils.isNotEmpty(clusterId)) {
                    try {
                        if (this.createTopic(task, clusterId)) {
                            Topic topic = new Topic();
//                            topic.setClusterId(clusterId);
                            Cluster cluster = new Cluster();
                            cluster.setId(Long.parseLong(clusterId));
                            topic.setCluster(cluster);
                            topic.setComments(task.getComments());
                            topic.setPartition(task.getPartition());
                            topic.setReplication(task.getReplication());
                            topic.setName(task.getName());
                            topic.setTtl(TimeUtil.hoursTransToMiss(task.getTtl()));

                            topicRepository.save(topic);
                        }
                    } catch (Exception e) {
                        flag = false;

                        ClusterDto cluster = clusterService.findById(Long.parseLong(clusterId));
                        int index = e.getMessage().indexOf(":");
                        backCreateTopic(task);

                        obj.put("true", false);
                        obj.put("message", "cluster: " + cluster.getName() + "  create topic has error:"
                                + e.getMessage().substring(index + 1, e.getMessage().length()));
                        return obj;
                    }
                }
            }
            obj.put("true", flag);
            obj.put("message", "success");
        }
        return obj;
    }

    private Boolean createTopic(TaskDto task, String clusterId) throws Exception {
        if (!kafkaService.getKafkaUtil(clusterId).createTopicIfNotExists(task.getName(),
                task.getPartition(), task.getReplication(), (task.getTtl() * 3600 * 1000))) {
            return false;
        }
        return true;
    }

    private void backCreateTopic(TaskDto task) {
        try {
            String[] clusterIds = task.getClusterIds().split(",");
            for (String clusterId : clusterIds) {
                KafkaUtil kafkaUtil = kafkaService.getKafkaUtil(clusterId);
                if (kafkaUtil.checkExists(task.getName())) {
                    kafkaUtil.delete(task.getName());
                }
            }
            topicRepository.deleteByName(task.getName());
        } catch (Exception e) {
            // TODO: handle exception
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean batchInsert(Set<Topic> topics) {
        int flagCount = 0;
        for (Topic topic : topics) {
            this.create(topic);
            flagCount++;
        }
        return flagCount == topics.size();
    }

    @Override
    public boolean batchUpdate(Set<Topic> topics) {
        int flagCount = 0;
        for (Topic topic : topics) {
            this.update(topic);
            flagCount++;
        }
        return flagCount == topics.size();
    }
}
